package com.zte.claa.inficombo.csa.app.service.nbiot.impl;

import java.nio.charset.Charset;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import javax.script.Invocable;

import org.apache.mina.core.buffer.IoBuffer;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import com.zte.claa.inficombo.csa.app.controller.ret.CmccOneNetRptData;
import com.zte.claa.inficombo.csa.app.model.dev.DevBase;
import com.zte.claa.inficombo.csa.app.model.dev.DevRunState;
import com.zte.claa.inficombo.csa.app.model.dev.DevTypeModel;
import com.zte.claa.inficombo.csa.app.model.redis.AlarmData;
import com.zte.claa.inficombo.csa.app.model.redis.DevMeteData;
import com.zte.claa.inficombo.csa.app.model.redis.SysLog;
import com.zte.claa.inficombo.csa.app.model.version.DevVersionUpdateLog;
import com.zte.claa.inficombo.csa.app.service.csa.impl.BaseCacheManager;
import com.zte.claa.inficombo.csa.app.service.jsplugin.JSParam;
import com.zte.claa.inficombo.csa.app.service.nbiot.NBIotCmdAck;
import com.zte.claa.inficombo.csa.app.service.nbiot.NBIotDevDataRptService;
import com.zte.claa.inficombo.csa.app.service.nbiot.IotLocUpdataProcessService;
import com.zte.claa.inficombo.csa.app.service.redis.PublishDataBuilder;
import com.zte.claa.inficombo.csa.app.service.rptparse.AppCmdWrapper;
import com.zte.claa.inficombo.csa.app.service.rptparse.LocRawData;
import com.zte.claa.inficombo.csa.app.service.rptparse.MultiPackageData;
import com.zte.claa.inficombo.csa.app.service.rptparse.ParserContainer;
import com.zte.claa.inficombo.csa.app.service.rptparse.ext.claa.ConfDataRequest;
import com.zte.claa.inficombo.csa.app.util.JsonUtils;

@Service
public class NBIotDevDataRptServiceImpl implements NBIotDevDataRptService, InitializingBean, DisposableBean {
	// js插件解析updata返还map时的key名称
	public static final String UPD_KEY_METEDATA = "metedata";
	public static final String UPD_KEY_ALARM    = "alarm";
	public static final String UPD_KEY_ACKCMD   = "ackcmd";
	public static final String UPD_KEY_MULTIPKG = "multipkgdata";
	public static final String UPD_KEY_CONFREQ  = "claaconfreq";
	public static final String UPD_KEY_METEDATALIST = "metedatalist";
	public static final String UPD_KEY_ALARMLIST    = "alarmlist";
	public static final String UPD_KEY_LOCRAWDATA   = "locrawdata";
	public static final String UPD_KEY_DEVVERLOG    = "verupdatedata";
	public static final String UPD_KEY_GSPCURVER    = "devcurversion";
	
	// NB设备fntype
	public static final int NB_DEV_FNTYPE = 4096;
	public static final int NB_DEV_TCPT_TRANSFER_FNTYPE = 4097;
	
	// 日志记录
	private final Logger logger = LoggerFactory.getLogger(NBIotDevDataRptServiceImpl.class);		
	
	@Autowired
	private BaseCacheManager cacheManager;
	
	@Autowired
	private ParserContainer container;
	
	@Autowired
	private NBIotCmccCmdExecService nbCmccCmdService;
	
	@Autowired
	private NBIotCtccCmdExecService nbCtccCmdService;
	
	@Autowired
	private NBIotCuccCmdExecService nbCuccCmdService;
	
	@Autowired
	private IotLocUpdataProcessService cmNbLocParseService;
	
	/**
	 * 持久化线程最大数目.
	 */
	@Value("${csa.cmcc.onenet.rpt.threadnum}")
	private int threadNum = 3;
	
	/**
	 * 线程池队列最大长度.
	 */
	@Value("${csa.cmcc.onenet.rpt.queuesize}")
	private int maxqueuesize = 1000;
	
	// 数据持久化和发布处理线程池...
	private ExecutorService execSvr;
	
	@Override
	public void afterPropertiesSet() throws Exception {
		this.execSvr = new ThreadPoolExecutor(threadNum, threadNum+3, 300, TimeUnit.MILLISECONDS, 
				new LinkedBlockingQueue<Runnable>(maxqueuesize));	
	}

	@Override
	public void destroy() throws Exception {
		if(this.execSvr != null){
			this.execSvr.shutdown();
			this.execSvr = null;
		}
	}

	/**
	 * 中移动NBIOT数据上报...
	 */
	@Override
	public boolean submitCmccOneNetRptData(CmccOneNetRptData rptData) {
		try{
			this.execSvr.execute(new Runnable(){
				@Override
				public void run() {
					processCmccOneNetRptData(rptData);
				}			
			});
		}catch(Exception e){
			logger.error("submitCmccOneNetRptData failed: "+rptData, e);
			return false;
		}		
		return true;
	}
		
	// 电信运营商的NBIOT数据上报接口，待后续定义和添加...
	public boolean submitCtccAepRptData(String rptJsonData) {
		logger.info("submitCtccAepRptData: {}", rptJsonData);
		try{
			this.processCtccAepRptData(rptJsonData);
			return true;
		}catch(Exception e){
			logger.error("submitCtccAepRptData failed: data={}", rptJsonData, e);
		}
		return false;
	}
			
	// 联通运营商的NBIOT数据上报接口，待后续定义和添加...
	public boolean submitCuccAepRptData(String rptJsonData){
		logger.info("submitCuccAepRptData: {}", rptJsonData);
		return true;
	}
		
	/**
	 * CMCC OneNet的测量指标数据处理: deveui=devId
	 * 
	 * 1）查询缓存，补充projecteid、devtype等类型；
	 * 2）调用devtype对应的js插件进行解析；
	 * 3）数据持久化保存；
	 * 4）数据发布；
	 * 
	 * @param CmccOneNetRptData
	 */
	@SuppressWarnings("unchecked")
	private void processCmccOneNetRptData(CmccOneNetRptData rptData) {
		if(rptData == null || rptData.getDev_id() == null || rptData.getDev_id().equals("")){
			logger.warn("DevId is null!");
			return;
		}
		String devEui = rptData.getDev_id(); 
		DevBase dev = this.cacheManager.getDev(devEui);
		if(dev == null){
			logger.warn("Dev instance NOT exist and data is discarded: {}, data={}", devEui, rptData);
			return;
		}
		int msgType = rptData.getType();
		// 终端状态数据上报.
		if(msgType == CmccOneNetRptData.MSG_TYPE_DEVST){
			int status = rptData.getStatus();
			if(status == CmccOneNetRptData.STATUS_OFF){
				this.setOnLine(devEui, DevBase.RUN_STATE_OFFLINE);
			} else {
				this.setOnLine(devEui, DevBase.RUN_STATE_ONLINE);
			}
			return;
		}
		else if(msgType == CmccOneNetRptData.MSG_TYPE_CPDATA){
			// 解析value字段.
			Object valObj = rptData.getValue();
			if(valObj == null){
				this.setOnLine(devEui, DevBase.RUN_STATE_ONLINE);	
				return;
			}
			IoBuffer buf = IoBuffer.allocate(48, false).setAutoExpand(true);
			if(ArrayList.class.isInstance(valObj)){
				//ASCII码字符数组.
				ArrayList<Integer> bsList = (ArrayList<Integer>)valObj;
				for(Integer bi : bsList){
					buf.put(bi.byteValue());
				}
			} 
			else if(Integer.class.isInstance(valObj)){
				buf.put(((Integer)valObj).byteValue());
			}
			else {
				byte[] bs = valObj.toString().getBytes(Charset.forName("ascii"));
				buf.put(bs);
			}
			buf.flip();
			byte[] bsArr = buf.array();
			int len = (bsArr.length < buf.limit()) ? bsArr.length : buf.limit();
			byte[] tmpBs = new byte[len];
			for(int i=0; i<len; i++){
				tmpBs[i] = bsArr[i];
			}
			buf.free();
			long rptTime = rptData.getAt();
			try{
				this.parseUpRptData(dev, rptTime, tmpBs);
			}catch(Exception e){
				logger.error("parseUpRptData for dev {} failed!", dev.getDevEui(), e);
			}
		} 
		else if(msgType == CmccOneNetRptData.MSG_TYPE_CMDACK){
			logger.info("[CMCC]Receive cmd ack message: "+rptData);
			// 将rptData记录到系统日志表，备查...
			if(this.container.getLogPublisher() != null && rptData != null){
				SysLog log = PublishDataBuilder.buildSysLog("CmccOneNet异步响应上报",
						SysLog.OP_RET_OK, 
						"CSA适配代理通讯框架", 
						dev.getDevName()+"("+dev.getDevEui()+")", 
						rptData.getOriginMsg());
				this.container.getLogPublisher().publishMessage(JsonUtils.toJson(log));
			}
		}
		this.setOnLine(devEui, DevBase.RUN_STATE_ONLINE);	
	}
	
	// 设置为在线/离线... DevBase.RUN_STATE_ONLINE | DevBase.RUN_STATE_OFFLINE
	private void setOnLine(String devEui, byte st){
		// 添加到dev-run-state队列中，更新dev的runstate=online|offline...
		DevRunState runState = new DevRunState();
		runState.setDevEUI(devEui);
		runState.setRunState(st);
		this.container.addDevRunStateData(runState);
	}
	
	/**
	 * 解析电信天翼AEP平台上报的消息
	 */
	private void processCtccAepRptData(String rptJsonData) {
		if(rptJsonData == null || rptJsonData.equals("")){
			logger.warn("[processCtccAepRptData] data is null!");
			return;
		}
		JSONObject jsonData = new JSONObject(rptJsonData);
		String msgType = JsonUtils.getJsonAttrVal(jsonData, "messageType", null);
		if(null == msgType){
			logger.warn("para [messageType] is null!");
			return;
		}
		
		String devEui = JsonUtils.getJsonAttrVal(jsonData, "deviceId", null);
		if(devEui == null){
			logger.warn("para [deviceId] is null!");
			return;
		}
		DevBase dev = this.cacheManager.getDev(devEui);
		if(dev == null){
			logger.warn("Dev instance NOT exist in cache and discard: {}, data={}", devEui, rptJsonData);
			return;
		}
		// 1. messageType消息类型=dataReport: 设备数据上报.
		if("dataReport".equalsIgnoreCase(msgType)){
			String payload = JsonUtils.getJsonAttrVal(jsonData, "payload", null);
			if(payload == null){
				this.setOnLine(devEui, DevBase.RUN_STATE_ONLINE);	
				return;
			}
			byte[] tmpBs = payload.getBytes(Charset.forName("UTF-8"));
			String rptTimeVal = JsonUtils.getJsonAttrVal(jsonData, "timestamp", "0");
			long rptTime = (rptTimeVal.equals("0")) ? System.currentTimeMillis() : Long.parseLong(rptTimeVal);
			try{
				this.parseUpRptData(dev, rptTime, tmpBs);
			}catch(Exception e){
				logger.error("[CTCC]parseUpRptData for dev {} failed!", dev.getDevEui(), e);
			}
		} 
		// 2.messageType消息类型=eventReport，告警事件上报.
		else if("eventReport".equalsIgnoreCase(msgType)){
			// 这里重新构造对象，方便js插件区分和解析.
			JSONObject alarmJson = new JSONObject();
			alarmJson.put("alarmevent", "1");			
			String eventType = JsonUtils.getJsonAttrVal(jsonData, "eventType", "1");
			// 事件类型，信息:1,警告:2,故障:3
			alarmJson.put("eventType", eventType);
			alarmJson.put("eventContent", JsonUtils.getSubObject(jsonData, "eventContent")); 
			byte[] tmpBs = alarmJson.toString().getBytes(Charset.forName("UTF-8"));
			String rptTimeVal = JsonUtils.getJsonAttrVal(jsonData, "timestamp", "0");
			long rptTime = (rptTimeVal.equals("0")) ? System.currentTimeMillis() : Long.parseLong(rptTimeVal);
			try{
				this.parseUpRptData(dev, rptTime, tmpBs);
			}catch(Exception e){
				logger.error("[CTCC]parseUpRptData-alarm for dev {} failed!", dev.getDevEui(), e);
			}
		}
		// 3. 终端状态上报: eventType	上线：1，下线：0; 
		else if("deviceOnlineOfflineReport".equalsIgnoreCase(msgType)){
			String eventType = JsonUtils.getJsonAttrVal(jsonData, "eventType", "0");
			int status = Integer.parseInt(eventType);
			if(status == 0){
				this.setOnLine(devEui, DevBase.RUN_STATE_OFFLINE);
			} else {
				this.setOnLine(devEui, DevBase.RUN_STATE_ONLINE);
			}
			return;
		}
		// 4. messageType消息类型=commandResponse (命令响应上报）
		else if("commandResponse".equalsIgnoreCase(msgType)){ 
			logger.info("[CTCC]Receive cmd ack message: devid={}", devEui);
			// 将rptData记录到系统日志表，备查...
			JSONObject subJson = JsonUtils.getSubObject(jsonData, "result");
			if(this.container.getLogPublisher() != null && subJson != null){
				SysLog log = PublishDataBuilder.buildSysLog("电信天翼AEP上报命令响应",
						SysLog.OP_RET_OK, 
						"CSA适配代理通讯框架", 
						dev.getDevName()+"("+dev.getDevEui()+")", 
						subJson.toString());
				this.container.getLogPublisher().publishMessage(JsonUtils.toJson(log));
			}
		}
		else {
			logger.info("[CTCC]other message type: {}", msgType);
		}
		this.setOnLine(devEui, DevBase.RUN_STATE_ONLINE);	
	}
	
	/**
	 * 上报数据默认解析方法
	 * 
	 * @param json
	 * @param dev
	 * @param svr
	 * @throws Exception
	 * @return 0:正常解析，不分包；1-分包；
	 * 
	 */
	@SuppressWarnings("unchecked")
	protected int parseUpRptData(DevBase dev, long rptTime, byte[] payload) throws Exception {
		String devTypeId = dev.getDevType();
		// 1.根据devtype获取解析插件（缓存）
		Invocable jsRunner = this.container.getJsPnpService().getJsPLugin(devTypeId);
		if(jsRunner == null){
			logger.error("[*!!*]Can not find the JS plugin for devtype="+devTypeId);
			return 0;
		}
		String portStr = "0";
		String devEui = dev.getDevEui();
		JSParam jsParam = new JSParam();
		jsParam.setDeveui(devEui);
		jsParam.setDevname(dev.getDevName());
		jsParam.setDevtypeid(devTypeId);
		jsParam.setDevport(portStr);
		jsParam.setAppeui(dev.getAppEui());
		jsParam.setProjectid(dev.getProjectId());
		jsParam.appendPara(JSParam.UPDATA_PARA_NAME, payload);
		// 如果涉及分包，则尝试设置分包中间结果.
		MultiPackageData mpd = this.container.getMultiPkgData(devEui);
		if(mpd != null){
			jsParam.appendPara(JSParam.MULTIPKG_PARA_NAME, mpd);
		}		
		// 模型附加配置信息
		String devExtData = dev.getDatajson();
		if (devExtData != null && !devExtData.trim().equals("")) {
			try {
				jsParam.setDevExtDatas(JsonUtils.jsonStrToHashMap(devExtData));
			} catch (Exception err) {
				logger.error("Get devextdata failed: " + devExtData + ", err=" + err.getMessage());
			}
		}
		// 设置持久化服务...
		jsParam.setPersistSvr(this.container.getPersisService());
		Object rptData = jsRunner.invokeFunction(JSParam.UPDATA_PARSE_FUNC, jsParam);
		// 情况1--解析结果无效；
		if(rptData == null){
			logger.info("[*!*]Invalid or nouse report data!");
			return 0;
		}
		// 情况2--只有alarm数据；	
		else if(AlarmData.class.isInstance(rptData)){
			logger.debug("Alarm data return from js plugin: "+rptData);
			this.parseAlarmData(rptTime, (AlarmData)rptData);
			return 0;
		}		
		// 情况3--只有metedata数据；
		else if(String.class.isInstance(rptData)){
			logger.debug("Mete data return from js plugin: "+rptData.toString());
			this.parseMeteData(dev, rptData.toString(), portStr, rptTime);
			return 0;
		}		
		// 情况4--同时有metedata、alarm数据、ackcmd（一种或几种兼有）；
		else if(HashMap.class.isInstance(rptData)){
			Map<String, Object> map = (HashMap<String, Object>)rptData;
			// 4.1单条告警...
			Object ao = map.get(UPD_KEY_ALARM);
			if (ao != null && AlarmData.class.isInstance(ao)) {
				logger.debug("Alarm data return from js plugin: " + ao);
				try {
					this.parseAlarmData(rptTime, (AlarmData) ao);
				} catch (Exception ex) {
					logger.error("parseAlarmData failed: " + ao, ex);
				}
			}
			// 4.2告警列表...
			Object aoList = map.get(UPD_KEY_ALARMLIST);
			if (aoList != null && ArrayList.class.isInstance(aoList)) {
				logger.debug("Alarm data list return from js plugin: " + aoList);
				ArrayList<AlarmData> alarmList = (ArrayList<AlarmData>) aoList;
				for (AlarmData alarm : alarmList) {
					try {
						this.parseAlarmData(rptTime, alarm);
					} catch (Exception ex) {
						logger.error("parseAlarmData failed: " + alarm, ex);
					}
				}
			}		
			// 4.3单条mdata（仅限于模型字段的json字符串）...
			Object mo = map.get(UPD_KEY_METEDATA);
			if (mo != null && String.class.isInstance(mo)) {
				logger.debug("Mete data return from js plugin: " + mo);
				try {
					this.parseMeteData(dev, mo.toString(), portStr, rptTime);
				} catch (Exception ex) {
					logger.error("parseMeteData failed: " + mo, ex);
				}
			}
			// 4.4mdata列表（仅限于模型字段的json字符串）...
			Object moList = map.get(UPD_KEY_METEDATALIST);
			if (moList != null && ArrayList.class.isInstance(moList)) {
				logger.debug("Mete data list return from js plugin: " + moList);
				ArrayList<String> mdataList = (ArrayList<String>) moList;
				this.parseMeteDataByBatch(dev, mdataList, portStr, rptTime);
			}
			// 4.5个别设备需要回复响应确认...
			Object ackCmd = map.get(UPD_KEY_ACKCMD);
			if (ackCmd != null && AppCmdWrapper.class.isInstance(ackCmd)) {
				logger.debug("Ack command return from js plugin: " + ackCmd);
				try {
					AppCmdWrapper cmdWrapper = (AppCmdWrapper)ackCmd;
					// send cmd in async way...
					this.container.sendDevCtrlCmd(cmdWrapper);
				} catch (Exception ex) {
					logger.error("Send ack cmd failed: " + ackCmd, ex);
				}
			}
			// 4.6 claa自研设备，初次入网时请求下发配置参数...
			Object confDataReqObj = map.get(UPD_KEY_CONFREQ);
			if (confDataReqObj != null && ConfDataRequest.class.isInstance(confDataReqObj)){
				logger.debug("claa dev req cmd conf data when first join: "+confDataReqObj);
				try{
					ConfDataRequest req = (ConfDataRequest)confDataReqObj;
					req.setDevtype(devTypeId);
					this.container.getDevConfDataSenderService().submitConfDataRequest(req);
				}catch(Exception ex){
					logger.error("submitConfDataRequest failed: "+confDataReqObj, ex);
				}
			}
			// 4.7 自研融合定位终端数据
			Object ldo = map.get(UPD_KEY_LOCRAWDATA);
			if (ldo != null && LocRawData.class.isInstance(ldo)) {
				logger.debug("Claa loc raw data return from js plugin: " + ldo);
				try {
					this.parseClaaLocRawData((LocRawData)ldo, dev);
				} catch (Exception ex) {
					logger.error("parseClaaLocRawData failed: " + ldo, ex);
				}
			}
			// 4.8 设备升级消息上报：包括升级准备命令响应、升级结果响应等  
			Object verupdateLog = map.get(UPD_KEY_DEVVERLOG);
			if (verupdateLog != null && DevVersionUpdateLog.class.isInstance(verupdateLog)) {
				logger.debug("Dev version update rsp return from js plugin: " + verupdateLog);
				try {
					this.processDevVerUpdateData((DevVersionUpdateLog)verupdateLog, dev);
				} catch (Exception ex) {
					logger.error("processDevVerUpdateData failed: " + verupdateLog, ex);
				}
			}			
			// 4.9 个别涉及分包情况.
			Object multiPkgObj = map.get(UPD_KEY_MULTIPKG);
			if(multiPkgObj != null && MultiPackageData.class.isInstance(multiPkgObj)){
				logger.debug("Multipackage data return from js plugin: " + multiPkgObj);
				try {
					MultiPackageData tmpMpd = (MultiPackageData)multiPkgObj;
					tmpMpd.setAppeui(dev.getAppEui());
					tmpMpd.setDeveui(devEui);
					tmpMpd.setDevName(dev.getDevName());
					tmpMpd.setDevtype(devTypeId);
					tmpMpd.setPortStr(portStr);
					tmpMpd.setProjectid(dev.getProjectId());					
					this.processMultiPkgData(tmpMpd, jsRunner, dev, portStr, rptTime);
				} catch (Exception ex) {
					logger.error("Multipackage data process failed: " + multiPkgObj, ex);
				}
				return 1;
			}
		}
		// 单个分包情况，JS插件基本上不会这么返回.
		else if (MultiPackageData.class.isInstance(rptData)){
			logger.debug("Multi-package data return from js plugin: "+rptData.toString());
			MultiPackageData tmpMpd = (MultiPackageData)rptData;
			tmpMpd.setAppeui(dev.getAppEui());
			tmpMpd.setDeveui(devEui);
			tmpMpd.setDevName(dev.getDevName());
			tmpMpd.setDevtype(devTypeId);
			tmpMpd.setPortStr(portStr);
			tmpMpd.setProjectid(dev.getProjectId());	
			this.processMultiPkgData(tmpMpd, jsRunner, dev, portStr, rptTime);
			return 1;
		}
		// 自研设备初次入网会请求下发参数.
		else if (ConfDataRequest.class.isInstance(rptData)){
			logger.debug("claa dev req cmd conf data when first join: "+rptData.toString());
			ConfDataRequest req = (ConfDataRequest)rptData;
			req.setDevtype(devTypeId);
			this.container.getDevConfDataSenderService().submitConfDataRequest(req);
			return 0;
		}	
		else if (LocRawData.class.isInstance(rptData)){
			logger.debug("claa loc raw data rpt: "+rptData.toString());
			this.parseClaaLocRawData((LocRawData)rptData, dev);
			return 0;
		}
		else if (DevVersionUpdateLog.class.isInstance(rptData)){
			logger.debug("claa dev version udpate data rpt: "+rptData.toString());
			this.processDevVerUpdateData((DevVersionUpdateLog)rptData, dev);
			return 0;
		}
		else {
			logger.warn("[*!*]Result data is invalid from js-plugin for devtype: "+devTypeId+", data="+rptData);
		}
		return 0;
	}
	
	/**
	 * 克拉自研CM-NB定位设备的上报数据解析.
	 * @param locData
	 */
	protected void parseClaaLocRawData(LocRawData locData, final DevBase dev) {
		this.cmNbLocParseService.parseClaaLocRawData(locData, dev);
	}
	
	/**
	 * 自研设备升级响应消息处理.
	 * @param verUpdateLog
	 * @param dev
	 */
	protected void processDevVerUpdateData(DevVersionUpdateLog verUpdateLog, final DevBase dev) {
		try{
			verUpdateLog.setAppeui(dev.getAppEui());			
			verUpdateLog.setProjectid(dev.getProjectId());
			verUpdateLog.setDevtype(dev.getDevType());
			if(verUpdateLog.getDeveui() == null){
				verUpdateLog.setDeveui(dev.getDevEui());
			}
			int rc = this.container.getPersisService().modifyDevVersionUpdateLog(verUpdateLog);
			logger.info("[^_^]modifyDevVersionUpdateLog ok: rc="+rc+", log="+verUpdateLog);
		}catch(Exception e){
			logger.error("modifyDevVersionUpdateLog failed: "+verUpdateLog, e);
		}
	}
	
	/**
	 * 发布告警数据。
	 * 注1：如果为告警恢复消息，且告警码alarmcode为"",则需要告警接收方根据当前deveui对应的告警情况作补充设置对应的告警码。
	 * 注2：告警接收端进行告警持久化或更新（根据告警缓存数据进行告警码设置，并保证告警数据的唯一性，CSA可能多实例部署）；
	 * @param alarm
	 */
	protected void parseAlarmData(long rptTime, AlarmData alarm){
		alarm.setAlarmtime(new Timestamp(rptTime));
		logger.debug("Publish alarm: "+alarm.toString());
		String alarmJson = JsonUtils.toJson(alarm);
		this.container.getAlarmPublisher().publishMessage(alarmJson);
		logger.info("[*^_^*]Publish alarm message ok: "+alarmJson);	
	}
	
	/**
	 * 批量发布mete数据
	 * 
	 * @param dev
	 * @param rptData
	 * @param devPort
	 * @throws Exception
	 */
	protected void parseMeteDataByBatch(DevBase dev, List<String> rptDataList, String devPort, long rptTime) throws Exception {
		String devTypeId = dev.getDevType();
		DevTypeModel dtm = this.container.getCacheManager().getDevTypeModel(devTypeId);
		if(dtm == null || dtm.getDmList() == null){
			logger.error("[*!!*]Can not get datamodel for devtype="+devTypeId);
			return;
		}
		List<DevMeteData> dataList = new ArrayList<DevMeteData>();
		for (String rptData : rptDataList) {
			Map<String, Object> mdata = JsonUtils.jsonStrToHashMap(rptData.toString());
			DevMeteData meteData = new DevMeteData();
			meteData.setAppeui(dev.getAppEui());
			meteData.setDeveui(dev.getDevEui());
			meteData.setDevname(dev.getDevName());
			meteData.setDevtype(devTypeId);
			meteData.setPort(devPort);
			meteData.setProjectid(dev.getProjectId());
			meteData.setMdata(mdata);
			meteData.setRpttime(new Timestamp(rptTime));
			dataList.add(meteData);
		}
		// 检测是否类似北秦电气火灾设备，需要二次合并数据记录.
		if(this.container.getMeteDataRedealService().needRedeal(devTypeId)){
			this.container.getMeteDataRedealService().submitMetedatas(devTypeId, dataList);
			return;
		}		
		boolean isBatchSaveOk = true;
		try{
			this.container.getPersisService().saveDevMeteDataByBatch(dataList, dtm.getBase().getTbname());
			isBatchSaveOk = true;
		}catch(Exception e){
			isBatchSaveOk = false;
			logger.error("saveDevMeteDataByBatch failed: "+dtm.getBase().getTbname(), e);
		}
		for(DevMeteData meteData : dataList){
			try{
				// 防止批量保存时个别异常导致全部入库异常.
				if(isBatchSaveOk == false){
					this.container.getPersisService().saveDevMeteData(meteData, dtm.getBase().getTbname());
				}
				// 构造广播数据，进行redis发布...
				String pubMsg = JsonUtils.toJson(meteData);
				this.container.getUpdataPublisher().publishMessage(pubMsg);
				logger.info("[*^_^*]Publish metedata message ok: "+pubMsg);
			}catch(Exception ex){
				logger.error("saveDevMeteData failed: "+dtm.getBase().getTbname(), ex);
			}
		}		
	}
	
	/**
	 * 发布mete数据
	 * 
	 * @param dev
	 * @param rptData
	 * @param devPort
	 * @throws Exception
	 */
	protected void parseMeteData(DevBase dev, String rptData, String devPort, long rptTime) throws Exception {
		String devTypeId = dev.getDevType();
		DevTypeModel dtm = this.container.getCacheManager().getDevTypeModel(devTypeId);
		if(dtm == null || dtm.getDmList() == null){
			logger.error("[*!!*]Can not get datamodel for devtype="+devTypeId);
			return;
		}
		Map<String, Object> mdata = JsonUtils.jsonStrToHashMap(rptData.toString());
		DevMeteData meteData = new DevMeteData();
		meteData.setAppeui(dev.getAppEui());
		meteData.setDeveui(dev.getDevEui());
		meteData.setDevname(dev.getDevName());
		meteData.setDevtype(devTypeId);
		meteData.setPort(devPort);
		meteData.setProjectid(dev.getProjectId());
		meteData.setRpttime(new Timestamp(rptTime));
		meteData.setMdata(mdata);
		// 检测是否类似北秦电气火灾设备，需要二次合并数据记录.
		if(this.container.getMeteDataRedealService().needRedeal(devTypeId)){
			this.container.getMeteDataRedealService().submitMetedata(meteData);
			return;
		}
		// 持久化数据.
		this.container.getPersisService().saveDevMeteData(meteData, dtm.getBase().getTbname());
		// 5.构造广播数据，进行redis发布...
		String pubMsg = JsonUtils.toJson(meteData);
		this.container.getUpdataPublisher().publishMessage(pubMsg);
		logger.info("[*^_^*]Publish metedata message ok: "+pubMsg);
	}
	
	/**
	 * 一、解析框架部分的主要逻辑：
	 * 1、如果当前MPD不空，则与返回结果合并(调用container.merge方法.)；
	 * 2、如果MPD.isDataOk=yes，isParsed=yes,confirm=yes，则此时JS已经做了metedata解析和ackcmd下发；MPD从缓存删除；调外部扩展接口；
	 * 3、如果MPD.isDataOk=yes，isParsed=yes,confirm=NO，  则此时JS已经做了metedata解析，框架侧走metedata解析流程；MPD入缓存；
	 * 4、如果MPD.isDataOk=yes，isParsed=no, confirm=NO，  则此时JS未做metedata解析，触发二次解析，设置isparsed=yes；MPD入缓存；
	 * 5、如果MPD.isDataOk=yes，isParsed=no, confirm=yes，则此时JS未做metedata解析，触发二次解析，设置isparsed=yes；MPD入缓存；
	 * 6、如果MPD.isDataOk=NO，confirm=yes，则此时JS已经做了ackcmd下发，框架侧走相应ACK流程；MPD入缓存；
	 * 7、如果MPD.isDataOk=NO，confirm=NO，此时MPD入缓存，等待后续包；
	 * 
	 * 二、相应地JS插件内部逻辑：（注意确认包序号是从1开始还是从0开始，JS-Framework约定从1开始）
	 * 1、接收原始byte[]流-bs，并尝试从JSParam中获取multipkgdata参数-mpd；
	 * 2、如果mpd参数为空，则创建一个mpd对象；
	 * 3、当前bs是普通分包数据包，则解析分包数据，确定包序号；
	 * 	   3.1 将包append到mpd中，注意：判断是否出现重复包，如果出现则将mpd重新初始化；
	 * 	   3.2 判断mpd是否dataok；
	 *     3.3 如果dataok，则调用buildData()解析出metedata（本次GNSS设备不需要，其他普通分包类型的情况需要），设置isparsed=yes；
	 *     3.4返回mpd及metedata（本次GNSS不需要），特别地，如果GNSS服务器只需要GNSS终端的净荷，还涉及data的特殊处理.
	 * 4、当前bs是分包上行校验包（分包发送完毕后的确认消息）；
	 *	   4.1 设置mpd的confirmed值=1；
	 *	   4.2 获取确认位图（getBitmap）及构造ackcmd，使用全1模式（ignoreabsentpkg=true）；
	 *	   4.3返回mpd和ackcmd；
	 * 5、新增二次解析方法入口： UPDATA_MULTIPKG_REPARSE_FUNC = "funcmultipkgreparse";
	 * 	   5.1框架针对分包的类型调用，入参是mpd；
	 *     5.2调用buildData()，得到完整码流，解析码流返回metedata（可选）和mpd（设置isparsed=yes）；
	 */
	protected void processMultiPkgData(MultiPackageData mpd, Invocable jsRunner, DevBase dev, String portStr, long rptTime) throws Exception {
		if(mpd == null){
			return;
		}
		MultiPackageData mergeData = this.container.mergePackageData(mpd);
		// 加锁互斥，避免重复处理.
		synchronized (mergeData) {
			boolean isConfirmed = (mergeData.getIsConfirmed() == MultiPackageData.CONFIRMED_YES);
			boolean isDataOk = mergeData.isDataOk();
			boolean isParsed = (mergeData.getIsParsed() == MultiPackageData.PARSED_YES);
			if (isDataOk && isParsed && isConfirmed) {
				this.container.removeMultiPkgData(mpd.getDeveui());
				logger.info("[1].remove MultiPkgData: deveui="+mpd.getDeveui());
			} else if (isDataOk && !isParsed) {
				// 首先设置解析标记.
				mergeData.setIsParsed(MultiPackageData.PARSED_YES);
				this.container.setMultiPkgData(mergeData);
				// 分包收集完整后进行二次解析
				this.reParseMultiPkgData(mergeData, jsRunner, dev, portStr, rptTime);
				// 解析结束，删除组包对象.
				this.container.removeMultiPkgData(mpd.getDeveui());
				logger.info("[2].finish MultiPkgData: deveui="+mpd.getDeveui());
			} else if (isDataOk && isParsed && !isConfirmed) {
				this.container.setMultiPkgData(mergeData);
				logger.debug("[3].reParseMultiPkgData: "+mpd);
			} else if (!isDataOk && isConfirmed) {
				this.container.setMultiPkgData(mergeData);
				logger.debug("[4].reParseMultiPkgData: "+mpd);
			} else if (!isDataOk && !isConfirmed) {
				this.container.setMultiPkgData(mergeData);
				logger.debug("[5].reParseMultiPkgData: "+mpd);
			}
		}
	}

	/**
	 * 分包收集完整后进行补充二次解析
	 * 
	 * @param mpd
	 * @param jsRunner
	 */
	@SuppressWarnings("unchecked")
	protected void reParseMultiPkgData(MultiPackageData mpd, Invocable jsRunner, DevBase dev, String portStr, long rptTime) throws Exception {
		// 设置可能的扩展配置数据，供js插件访问.
		this.setExtDataToMultiPkg(mpd, dev);
		// 开始二次解析.
		Object rptData = jsRunner.invokeFunction(JSParam.UPDATA_MULTIPKG_REPARSE_FUNC, mpd);
		// 情况1--解析结果无效；
		if(rptData == null){
			logger.info("[*!*]MPD-Reparse: Invalid or nouse report data!");
			return;
		}
		// 情况2--只有alarm数据；	
		else if(AlarmData.class.isInstance(rptData)){
			logger.debug("MPD-Reparse: Alarm data return from js plugin: "+rptData);
			this.parseAlarmData(rptTime, (AlarmData)rptData);
			return;
		}		
		// 情况3--只有metedata数据；
		else if(String.class.isInstance(rptData)){
			logger.debug("MPD-Reparse: Mete data return from js plugin: "+rptData.toString());
			this.parseMeteData(dev, rptData.toString(), portStr, rptTime);
			return;
		}		
		// 情况3.5 自研设备初次入网会请求下发参数.
		else if (ConfDataRequest.class.isInstance(rptData)){
			logger.debug("claa dev req cmd conf data when first join: "+rptData.toString());
			ConfDataRequest req = (ConfDataRequest)rptData;
			req.setDevtype(dev.getDevType());
			this.container.getDevConfDataSenderService().submitConfDataRequest(req);
		}	
		// 情况4--同时有metedata、alarm数据、ackcmd（一种或几种兼有）；
		else if(HashMap.class.isInstance(rptData)){
			Map<String, Object> map = (HashMap<String, Object>)rptData;
			// 4.1单条告警...
			Object ao = map.get(UPD_KEY_ALARM);
			if (ao != null && AlarmData.class.isInstance(ao)) {
				logger.debug("Alarm data return from js plugin: " + ao);
				try {
					this.parseAlarmData(rptTime, (AlarmData) ao);
				} catch (Exception ex) {
					logger.error("parseAlarmData failed: " + ao, ex);
				}
			}
			// 4.2告警列表...
			Object aoList = map.get(UPD_KEY_ALARMLIST);
			if (aoList != null && ArrayList.class.isInstance(aoList)) {
				logger.debug("Alarm data list return from js plugin: " + aoList);
				ArrayList<AlarmData> alarmList = (ArrayList<AlarmData>) aoList;
				for (AlarmData alarm : alarmList) {
					try {
						this.parseAlarmData(rptTime, alarm);
					} catch (Exception ex) {
						logger.error("parseAlarmData failed: " + alarm, ex);
					}
				}
			}		
			// 4.3单条mdata（仅限于模型字段的json字符串）...
			Object mo = map.get(UPD_KEY_METEDATA);
			if (mo != null && String.class.isInstance(mo)) {
				logger.debug("Mete data return from js plugin: " + mo);
				try {
					this.parseMeteData(dev, mo.toString(), portStr, rptTime);
				} catch (Exception ex) {
					logger.error("parseMeteData failed: " + mo, ex);
				}
			}
			// 4.4mdata列表（仅限于模型字段的json字符串）...
			Object moList = map.get(UPD_KEY_METEDATALIST);
			if (moList != null && ArrayList.class.isInstance(moList)) {
				logger.debug("Mete data list return from js plugin: " + moList);
				ArrayList<String> mdataList = (ArrayList<String>) moList;
				this.parseMeteDataByBatch(dev, mdataList, portStr, rptTime);
			}
			// 4.5个别设备需要回复响应确认...
			Object ackCmd = map.get(UPD_KEY_ACKCMD);
			if (ackCmd != null && AppCmdWrapper.class.isInstance(ackCmd)) {
				logger.debug("Ack command return from js plugin: " + ackCmd);
				try {
					AppCmdWrapper cmdWrapper = (AppCmdWrapper)ackCmd;
					// send cmd in async way...
					this.container.sendDevCtrlCmd(cmdWrapper);
				} catch (Exception ex) {
					logger.error("Send ack cmd failed: " + ackCmd, ex);
				}
			}
			// 4.6 claa自研设备，初次入网时请求下发配置参数...
			Object confDataReqObj = map.get(UPD_KEY_CONFREQ);
			if (confDataReqObj != null && ConfDataRequest.class.isInstance(confDataReqObj)){
				logger.debug("claa dev req cmd conf data when first join: "+confDataReqObj);
				try{
					ConfDataRequest req = (ConfDataRequest)rptData;
					req.setDevtype(dev.getDevType());
					this.container.getDevConfDataSenderService().submitConfDataRequest(req);
				}catch(Exception ex){
					logger.error("submitConfDataRequest failed: "+confDataReqObj, ex);
				}
			}
			// 4.8 自研融合定位终端数据
			Object ldo = map.get(UPD_KEY_LOCRAWDATA);
			if (ldo != null && LocRawData.class.isInstance(ldo)) {
				logger.debug("Claa loc raw data return from js plugin: " + ldo);
				try {
					this.parseClaaLocRawData((LocRawData)ldo, dev);
				} catch (Exception ex) {
					logger.error("parseClaaLocRawData failed: " + ldo, ex);
				}
			}
		}
		else if (LocRawData.class.isInstance(rptData)){
			logger.debug("[2]-claa loc raw data rpt: "+rptData.toString());
			this.parseClaaLocRawData((LocRawData)rptData, dev);
		}
		else {
			logger.warn("[*!*]Result data is invalid from js-plugin for devtype: "+dev.getDevType()+", data="+rptData);
		}
	}
	
	/**
	 * 分包时扩增配置数据的设置.
	 */
	protected void setExtDataToMultiPkg(MultiPackageData mpd, DevBase dev) {
		try {
			String devExtData = dev.getDatajson();
			if (devExtData == null || devExtData.trim().equals("")) {
				return;
			}
			mpd.setDevExtDatas(JsonUtils.jsonStrToHashMap(devExtData));
		} catch (Exception e) {
			logger.error("setExtDataToMultiPkg failed: " + dev + ", err=" + e.getMessage());
		}
	}
	
	/**
	 * 命令转换和下发.
	 * 1、区分cmcc、cucc、ctcc三类；
	 * 2、构造对象；
	 * 3、调用js插件构造args参数；
	 * 4、执行命令；
	 * 5、解析结果返回；
	 */
	public NBIotCmdAck execNBIotCmd(DevBase dev, String payload, Map<String, Object> extParaMap, boolean offlineMode) throws Exception {
		byte devFlag = dev.getDevflag();
		if(devFlag == DevBase.DEV_TPFLAG_NB_CMCC){
			return this.nbCmccCmdService.execCmccNBIotCmd(dev, payload, extParaMap, offlineMode);
		} 		
		else if(devFlag == DevBase.DEV_TPFLAG_NB_CUCC){
			return this.nbCuccCmdService.execCuccNBIotCmd(dev, payload, extParaMap, offlineMode);
		} 
		else if(devFlag == DevBase.DEV_TPFLAG_NB_CTCC){
			return this.nbCtccCmdService.execCtccNBIotCmd(dev, payload, extParaMap, offlineMode);
		} 	
		return new NBIotCmdAck(0, "Invalid devflag "+devFlag, "");
	}
	
}
